package com.zhang.spark_2.com.zhang.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @title:
 * @author: zhang
 * @date: 2022/2/17 09:47 
 */
object WC_aggregate {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);

    val rdd = sc.makeRDD(
      List(("a", 1), ("b", 2), ("a", 1), ("b", 2), ("a", 1), ("b", 2)), 2
    )

    val word: RDD[String] = rdd.map {
      case (word, num) => {
        (word + " ") * num
      }
    }.flatMap(_.split(" "))

    // todo 使用aggregate实现wordCount
    val wc: mutable.Map[String, Int] = word.aggregate(mutable.Map[String, Int]())(
      (map, word) => {
        val old: Int = map.getOrElse(word, 0)
        map.update(word, old + 1)
        map
      },
      (map1, map2) => {
        map2.foldLeft(map1) {
          (map1, kv) => {
            val word: String = kv._1
            val cnt: Int = kv._2
            val old: Int = map1.getOrElse(word, 0)
            map1.update(word, old + cnt)
            map1
          }
        }
      }
    )
    println(wc)

    sc.stop()
  }

}
